package groupManagementModule;

import exceptions.NameTakenException;
import exceptions.NotLeaderException;
import generalObjects.BaseClientContract;
import generalObjects.CassandraHandler;
import generalObjects.GroupInformation;
import generalObjects.GroupView;
import generalObjects.LayerAccess;

import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.server.ServerNotActiveException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

import javax.swing.event.ChangeListener;

import com.datastax.driver.core.ConsistencyLevel;
import communicationModule.MessageObject;
import communicationModule.SendResult;

/**
 * The top most layer of the GCom stack, this class is responsible for keeping
 * track of incoming/outgoing messages in the system. As well as already
 * received messages. This object is bound to the RMI registry by the Model. It
 * provides remote methods for sending messages to the client as well as joining
 * the group if the client is also the group leader.
 * 
 */
public class GMMLayerAccess implements LayerAccess, BaseClientContract {
    private LayerAccess lowerLayer;
    private final String[] nameservers;
    private GroupView gView;
    private boolean historyRet = false;
    private final Properties sessionProp;
    private final List<MessageObject> workQueue;
    private final ReentrantLock workQLock;
    private MessageHandler mHandler;
    private final ConcurrentHashMap<Long, Integer> receivedbuf;
    private CassandraHandler cHandler;
    private final ChangeListener groupStateListener;
    private final AtomicBoolean groupChanged = new AtomicBoolean(false);
    private HistoryRetriever histRetriever;
    private ConcurrentHashMap<String, Integer> stampclock;

    public GMMLayerAccess(final Properties sessionProp,
            final String[] nameservers, ChangeListener groupStateListener,
            final ConcurrentHashMap<Long, Integer> receivedbuf) {
        this.sessionProp = sessionProp;
        this.nameservers = nameservers;
        this.receivedbuf = receivedbuf;
        this.groupStateListener = groupStateListener;
        workQLock = new ReentrantLock();
        workQueue = new ArrayList<MessageObject>();
    }

    @Override
    public void isAlive() throws RemoteException {
        // used to check live status of a leader by server.
    }

    @Override
    public void send(final MessageObject msg) throws RemoteException {

        workQLock.lock();
        workQueue.add(msg);
        workQLock.unlock();
    }

    @Override
    public GroupInformation join(final BaseClientContract stub,
            final String nick) throws RemoteException, NameTakenException,
            NotLeaderException {
        GroupInformation gInfo;
        final String myNick = sessionProp.getProperty("nickname");
        final String order = sessionProp.getProperty("ordering");
        final String multicast = sessionProp.getProperty("multicast");
        if (!myNick.equals(gView.getLeader())) {
            throw new NotLeaderException(
                    "You can not call join on non-leader client");
        }
        if (gView.getView().containsKey(nick)) {
            throw new NameTakenException("This name is taken :" + nick);
        }

        // gView.appendMember(nick, stub, 0);
        if (!stampclock.containsKey(nick)) {
            gView.appendMember(nick, stub, 0);
        } else {
            gView.appendMember(nick, stub, stampclock.get(nick));
        }
        groupChanged.compareAndSet(false, true);
        gInfo = new GroupInformation(gView, order, multicast);
        sendUpdateGroupView(nick);
        return gInfo;
    }

    public void setGroupView(final GroupView gV) {
        gView = gV;
        mHandler = new MessageHandler(gView, sessionProp, nameservers,
                groupChanged);
    }

    public void setCassandraHandler(final CassandraHandler cHandler) {
        this.cHandler = cHandler;
    }

    @Override
    public void setLowerLayer(final LayerAccess layer) {
        lowerLayer = layer;
    }

    @Override
    public List<SendResult> process(final List<MessageObject> mOList)
            throws RemoteException, NotBoundException, ServerNotActiveException {

        final List<SendResult> delivery = mHandler.handleResult(lowerLayer
                .process(mOList));
        if (groupChanged.get()) {
            groupStateListener.stateChanged(null);
            groupChanged.compareAndSet(true, false);
        }
        // saving messages sent by myself to the Cassandra database.
        final String insertMsg = "INSERT INTO grouphistory "
                + "(groupname,id,header,data,sender,senderseqnr,vectorclock)"
                + " VALUES(?,?,?,?,?,?,?)";
        final String gName = sessionProp.getProperty("groupname");
        final String myNick = sessionProp.getProperty("nickname");
        if (delivery == null) {
            return delivery;
        }

        for (final SendResult del : delivery) {
            final MessageObject mObject = del.getMsg();
            if (mObject.getSender().equals(myNick)
                    && mObject.getHeader().equals("message")) {

                final long id = mObject.getId();
                final String header = mObject.getHeader();
                final String data = mObject.getContent();
                final String sender = mObject.getSender();
                ConcurrentHashMap<String, Integer> vClock;
                int seqNr;
                vClock = mObject.getGroupView().getVectorClock();
                seqNr = vClock.get(myNick);

                if (sessionProp.getProperty("ordering").equals("false")) {
                    // extend with ourself if missing.
                    if (!stampclock.containsKey(myNick)) {
                        stampclock.put(myNick, 0);
                    }
                    // fix for increment own clock unordered
                    int senderseqnr = stampclock.get(myNick);
                    senderseqnr++;
                    stampclock.put(myNick, senderseqnr);

                    for (final Map.Entry<String, Integer> ent : stampclock
                            .entrySet()) {

                        vClock.put(ent.getKey(), ent.getValue());
                    }
                } else {
                    for (final Map.Entry<String, Integer> ent : stampclock
                            .entrySet()) {
                        vClock.putIfAbsent(ent.getKey(), ent.getValue());
                    }
                }

                // extend the msg clock

                final String vClockJSON = cHandler.mapToJSONString(vClock);
                cHandler.executeStatement(insertMsg, ConsistencyLevel.QUORUM,
                        gName, id, header, data, sender, seqNr, vClockJSON);
            } else if (sessionProp.getProperty("ordering").equals("false")
                    && mObject.getHeader().equals("message")) {
                // extend with ourself if missing.

                // someone elses msg in unordered mode.

                final String nameofSender = mObject.getSender();
                if (!stampclock.containsKey(nameofSender)) {
                    stampclock.put(nameofSender, 0);
                }
                int senderseqnr = stampclock.get(nameofSender);
                senderseqnr++;
                stampclock.put(nameofSender, senderseqnr);

            }
        }

        if (sessionProp.getProperty("ordering").equals("true")) {
            for (final Map.Entry<String, Integer> ent : gView.getVectorClock()
                    .entrySet()) {
                final String memNick = ent.getKey();
                stampclock.put(memNick, ent.getValue());
            }
        }

        // stamp last seen vclock in DB.
        final String insertStamp = "INSERT INTO usersstamp "
                + "(groupname,nick,myseqnr,lastseenvclock) VALUES(?,?,?,?)";
        if (!stampclock.containsKey(myNick)) {
            stampclock.put(myNick, 0);
        }
        int myseqnr = stampclock.get(myNick);
        final String stampclockJSON = cHandler.mapToJSONString(stampclock);

        cHandler.executeStatement(insertStamp, ConsistencyLevel.QUORUM, gName,
                myNick, myseqnr, stampclockJSON);

        return delivery;
    }

    public List<SendResult> process() throws RemoteException,
            NotBoundException, ServerNotActiveException {
        if (!historyRet) {
            historyRet = true;

            histRetriever = new HistoryRetriever(cHandler,
                    sessionProp.getProperty("nickname"),
                    sessionProp.getProperty("groupname"), receivedbuf);
            final List<SendResult> history = histRetriever.retrieveMsgs();
            // sync clocks
            stampclock = histRetriever.getGroupClock();
            for (final Map.Entry<String, Integer> ent : gView.getVectorClock()
                    .entrySet()) {
                if (stampclock.containsKey(ent.getKey())) {
                    final int memgclock = stampclock.get(ent.getKey());
                    if (memgclock > ent.getValue()) {
                        ent.setValue(memgclock);
                    }
                }
            }

            return history;
        }
        boolean lockSucceded = false;
        ArrayList<MessageObject> msgs = null;
        lockSucceded = workQLock.tryLock();
        if (lockSucceded) {
            msgs = new ArrayList<MessageObject>();
            for (int i = 0; i < workQueue.size(); i++) {

                msgs.add(workQueue.remove(i));
            }
            workQLock.unlock();
        }
        return process(msgs);
    }

    private void sendUpdateGroupView(final String nick) throws RemoteException {
        final String myNick = sessionProp.getProperty("nickname");
        final MessageObject msg = new MessageObject(myNick, "update", nick,
                gView, "");
        gView.getView().get(myNick).send(msg);
    }

    @Override
    public GroupView getVectorClockBaseline() throws RemoteException {

        return gView;
    }
}
